package com.atguigu.bigdata.chapter09;

import com.atguigu.bigdata.bean.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * @Author lzc
 * @Date 2022/9/8 15:02
 */
public class Flink04_CEP_Combine_Pattern {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 2000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        // 1. 先有流: 流必须要设置事件时间和水印
        SingleOutputStreamOperator<WaterSensor> stream = env
            .readTextFile("input/sensor.txt")
            .map(line -> {
                String[] data = line.split(",");
                return new WaterSensor(
                    data[0],
                    Long.valueOf(data[1]),
                    Integer.valueOf(data[2])
                );
            })
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<WaterSensor>forMonotonousTimestamps()
                    .withTimestampAssigner((ws, ts) -> ws.getTs())
            );
        
        // 2. 定义规则(模式) 类似正则表达式
        Pattern<WaterSensor, WaterSensor> pattern = Pattern
            .<WaterSensor>begin("s1")
            .where(new SimpleCondition<WaterSensor>() {
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "sensor_1".equals(value.getId());
                }
            })
            //            .next("s2") // 严格连续(严格近邻)  依据的是事件时间
            //.notNext("s2") // 非严格近邻  一个sensor_1, 后面有数据, 但是不是sensor_2
//            .followedBy("s2") // 松散连续: sensor_1后面跟着sensor_2, 不管中间是否有其他数据
            
            // notFollowedBy 不能作为模式的最后一个部分
            .notFollowedBy("s2") // 非松散连续: 有个sensor_1, 后面跟着sensor_3, 但是中间不能有sensor_2, 这样的sensor_1和sensor_3满足模式
            .where(new SimpleCondition<WaterSensor>() {
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "sensor_2".equals(value.getId());
                }
            })
            .followedBy("s3")
            .where(new SimpleCondition<WaterSensor>() {
                @Override
                public boolean filter(WaterSensor value) throws Exception {
                    return "sensor_3".equals(value.getId());
                }
            });
        
        
        
        // 3. 把规则作用到流上, 得到一个模式流
        PatternStream<WaterSensor> ps = CEP.pattern(stream, pattern);
        
        // 4. 从模式流中取出匹配到的数据
        SingleOutputStreamOperator<String> result = ps
            .select(new PatternSelectFunction<WaterSensor, String>() {
                // map: key 模式的名字  value: 匹配成功一次 之后满足这个模式的所有数据
                @Override
                public String select(Map<String, List<WaterSensor>> map) throws Exception {
                    return map.toString();
                }
            });
        
        result.print();
        
        
        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
